package tasks.io.core.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.core.*;
import tasks.io.core.WriteCompletionHandler.WriteCompletionVo;
import tasks.io.core.thread.AbstractQueueTask;
import tasks.io.core.utils.IOUtils;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReentrantLock;


public class SendTask extends AbstractQueueTask<Packet> {
    private static final Logger log = LoggerFactory.getLogger(SendTask.class);
    private ChannelContext channelContext = null;
    private TioConfig tioConfig = null;
    private TioHandler tioHandler = null;


    public boolean canSend = true;

    public SendTask(ChannelContext channelContext, Executor executor) {
        super(executor);
        this.channelContext = channelContext;
        this.tioConfig = channelContext.tioConfig;
        this.tioHandler = tioConfig.getTioHandler();


        getQueue();
    }

    @Override
    public boolean add(Packet packet) {
        if (this.isCanceled()) {
            log.info("{}, 任务已经取消，{}添加到发送队列失败", channelContext, packet.logstr());
            return false;
        }


        {
            return queue.add(packet);
        }
    }

    /**
     * 清空消息队列
     */
    @Override
    public void clearQueue() {
        Packet p = null;
        //new ConcurrentLinkedQueue<>();
        ConcurrentLinkedQueue<Packet> forSendAfterSslHandshakeCompleted = null;
        while ((p = queue.poll()) != null) {
            try {
                channelContext.processAfterSent(p, false);
            } catch (Throwable e) {
                log.error(e.toString(), e);
            }
        }
    }

    private ByteBuffer getByteBuffer(Packet packet) {
        try {
            ByteBuffer byteBuffer = packet.getPreEncodedByteBuffer();
            if (byteBuffer == null) {
                byteBuffer = tioHandler.encode(packet, tioConfig, channelContext);
            }

            if (!byteBuffer.hasRemaining()) {
                byteBuffer.flip();
            }
            return byteBuffer;
        } catch (Exception e) {
            log.error(packet.logstr(), e);
            throw new RuntimeException(e);
        }
    }


    private static final int MAX_CAPACITY_MIN = TcpConst.MAX_DATA_LENGTH - 1024;    //减掉1024是尽量防止溢出的一小部分还分成一个tcp包发出
    private static final int MAX_CAPACITY_MAX = MAX_CAPACITY_MIN * 10;

    //	private int repeatCount = 0;

    @Override
    public void runTask() {
        if (queue.isEmpty()) {
            return;
        }

        int queueSize = queue.size();
        if (queueSize == 1) {
            //			System.out.println(1);
            Packet packet = queue.poll();
            if (packet != null) {
                sendPacket(packet);
            }
            return;
        }

        int listInitialCapacity = Math.min(queueSize, canSend ? 300 : 1000);

        Packet packet = null;
        List<Packet> packets = new ArrayList<>(listInitialCapacity);
        List<ByteBuffer> byteBuffers = new ArrayList<>(listInitialCapacity);
        //		int packetCount = 0;
        int allBytebufferCapacity = 0;
        Boolean needSslEncrypted = null;
        boolean sslChanged = false;
        while ((packet = queue.poll()) != null) {
            ByteBuffer byteBuffer = getByteBuffer(packet);

            packets.add(packet);
            byteBuffers.add(byteBuffer);
            //			packetCount++;
            allBytebufferCapacity += byteBuffer.limit();


            if ((canSend && allBytebufferCapacity >= MAX_CAPACITY_MIN) || (allBytebufferCapacity >= MAX_CAPACITY_MAX) || sslChanged) {
                break;
            }
        }


        if (allBytebufferCapacity == 0) {
            return;
        }
        ByteBuffer allByteBuffer = ByteBuffer.allocate(allBytebufferCapacity);
        for (ByteBuffer byteBuffer : byteBuffers) {
            allByteBuffer.put(byteBuffer);
        }

        allByteBuffer.flip();


        this.sendByteBuffer(allByteBuffer, packets);
        //		queueSize = msgQueue.size();
        //		if (queueSize > 0) {
        //			repeatCount++;
        //			if (repeatCount < 3) {
        //				runTask();
        //				return;
        //			}
        //		}
        //		repeatCount = 0;
    }

    public boolean sendPacket(Packet packet) {
        ByteBuffer byteBuffer = getByteBuffer(packet);


        sendByteBuffer(byteBuffer, packet);
        return true;
    }

    /**
     * @param byteBuffer
     * @param packets    Packet or List<Packet>
     * @author tanyaowu
     */
    public void sendByteBuffer(ByteBuffer byteBuffer, Object packets) {
        if (byteBuffer == null) {
            log.error("{},byteBuffer is null", channelContext);
            return;
        }

        if (!IOUtils.checkBeforeIO(channelContext)) {
            return;
        }

        //		if (!byteBuffer.hasRemaining()) {
        //			byteBuffer.flip();
        //		}

        ReentrantLock lock = channelContext.writeCompletionHandler.lock;
        lock.lock();
        try {
            canSend = false;
            WriteCompletionVo writeCompletionVo = new WriteCompletionVo(byteBuffer, packets);
            channelContext.asynchronousSocketChannel.write(byteBuffer, writeCompletionVo, channelContext.writeCompletionHandler);
            channelContext.writeCompletionHandler.condition.await();
        } catch (InterruptedException e) {
            log.error(e.toString(), e);
        } finally {
            lock.unlock();
        }
    }


    private final ConcurrentLinkedQueue<Packet> queue = new ConcurrentLinkedQueue<Packet>();

    @Override
    public ConcurrentLinkedQueue<Packet> getQueue() {

        return queue;
    }

}
